package com.bawei.flink.demo;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

/**
 * 输入
 * 10
 * 20
 * 15
 * 22
 * 31
 * 35
 * 77
 * 88
 * 99
 * 100
 * 10
 * 10
 * 10
 * 10
 * 10
 * 10
 * 10
 * 10
 * 10
 * 10
 *
 * 输出
 * 49.7
 * 10.0
 *
 */
public class GlobalWindowDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> lines = env.socketTextStream("cdh1", 9999);

        lines.windowAll(GlobalWindows.create()).trigger(new Trigger<String, GlobalWindow>() {

            ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("total", Integer.class);

            @Override
            public TriggerResult onElement(String element, long timestamp, GlobalWindow window, TriggerContext ctx) throws Exception {
                ValueState<Integer> total = ctx.getPartitionedState(stateDescriptor);
                if(total.value() == null) {
                    total.update(1);
                } else {
                    Integer i = total.value();
                    i += 1;
                    total.update(i);
                }
                if(total.value() >= 10) {
                    total.update(0);
                    return TriggerResult.FIRE_AND_PURGE;
                } else {
                    return TriggerResult.CONTINUE;
                }

            }

            @Override
            public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
                return TriggerResult.CONTINUE;
            }

            @Override
            public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
                return TriggerResult.CONTINUE;
            }

            @Override
            public void clear(GlobalWindow window, TriggerContext ctx) throws Exception {
                ctx.getPartitionedState(stateDescriptor).clear();
            }
        })
                .process(new ProcessAllWindowFunction<String, Double, GlobalWindow>() {
                    @Override
                    public void process(Context context, Iterable<String> elements, Collector<Double> out) throws Exception {
                        Integer sum = 0;
                        Integer count = 0;
                        for(String element : elements) {
                            Integer num = Integer.valueOf(element);
                            sum += num;
                            count ++;
                        }
                        double avg = sum.doubleValue() / count;
                        out.collect(avg);
                    }
                })
                .print();

        env.execute();
    }

}
